package com.eric.study.group;

import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.PostStop;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import com.eric.study.device.Device;
import com.eric.study.manager.DeviceManager;
import com.eric.study.message.Message;
import org.slf4j.Logger;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

/**
 * <p>
 * DeviceGroup
 *
 * <p>
 * change history:
 *
 * <pre>
 * date         defect       person          comments
 * --------------------------------------------------------------------------------
 * 2023/8/19    ********     Eric    create file.
 * </pre>
 *
 * @author Eric
 * @date 2023/8/19 15:05
 * @since 1.0.0
 */
public class DeviceGroup extends AbstractBehavior<DeviceGroup.GroupMessage> {

    /**
     * 组编号，一个 Actor 一个编号
     */
    private final String groupId;

    /**
     * 当前设备组下管理的所有设备 Actor 引用
     */
    private final Map<String, ActorRef<Device.DeviceMessage>> deviceIdToActor;

    private final Logger log;

    private DeviceGroup(ActorContext<GroupMessage> context, String groupId) {
        super(context);
        this.groupId = groupId;
        this.log = context.getLog();
        this.deviceIdToActor = new HashMap<>();

        log.info("DeviceGroup [{}] started. path: {}", groupId, context.getSelf());
    }

    public static Behavior<GroupMessage> create(String groupId) {
        return Behaviors.setup(context -> new DeviceGroup(context, groupId));
    }

    @Override
    public Receive<GroupMessage> createReceive() {
        return newReceiveBuilder()
                // 接收并处理 RequestTrackDevice 命令
                .onMessage(DeviceManager.RequestTrackDevice.class, this::onTrackDevice)
                // 接收并处理 Actor 异常停止命令
                .onMessage(DeviceTerminated.class, this::onTerminated)
                // 接收并处理查询设备列表命令
                .onMessage(DeviceManager.QueryDeviceList.class, this::onDeviceList)
                // 接收并处理查询所有设备温度命令
                .onMessage(DeviceManager.QueryAllTemperatures.class, this::onAllTemperatures)
                .onMessage(DeviceManager.StopDevice.class, this::stopDevice)
                .onSignal(PostStop.class, signal -> {
                    log.info("DeviceGroup [{}] stopped.", groupId);
                    return this;
                })
                .build();
    }

    private Behavior<GroupMessage> stopDevice(DeviceManager.StopDevice msg) {
        ActorRef<Device.DeviceMessage> device = deviceIdToActor.get(msg.deviceId());
        if (Objects.nonNull(device)) {
            device.tell(Device.Passivate.INSTANCE);
        }
        return this;
    }

    /**
     * 通过将仅与查询本身相关的临时状态保存在单独的参与者中，我们可以使 GroupActor 的实现非常简单。
     *
     * <p>它将所有内容委托给子角色，因此不必保留与其核心业务无关的状态。此外，多个查询现在可以彼此并行运行，事实上，可以根据需要运行多个查询。
     *
     * @author Eric
     */
    private DeviceGroup onAllTemperatures(DeviceManager.QueryAllTemperatures msg) {
        // 由于Java集合是可变的，我们希望避免在 Actor 之间共享它们(因为多个actor(线程)修改相同的可变数据结构是不安全的)
        // 在Java应用程序中，您可以自由地将您喜欢的不可变数据结构库与Akka一起使用!
        Map<String, ActorRef<Device.DeviceMessage>> deviceIdToActorCopy = new HashMap<>(this.deviceIdToActor);

        // 转发查询命令
        getContext().spawnAnonymous(DeviceGroupQuery.create(
                deviceIdToActorCopy,
                msg.requestId(),
                msg.replyTo(),
                Duration.ofSeconds(3)));

        return this;
    }

    private DeviceGroup onDeviceList(DeviceManager.QueryDeviceList msg) {
        msg.replyTo().tell(new DeviceManager.ReplyDeviceList(msg.requestId(), deviceIdToActor.keySet()));
        return this;
    }

    private DeviceGroup onTerminated(DeviceTerminated msg) {
        log.info("Device actor for [{}] has been terminated.", msg.deviceId());
        deviceIdToActor.remove(msg.deviceId());
        System.out.println("=> " + groupId + " Device: " + deviceIdToActor);
        return this;
    }


    private DeviceGroup onTrackDevice(DeviceManager.RequestTrackDevice msg) {
        if (this.groupId.equals(msg.groupId())) {
            ActorRef<Device.DeviceMessage> deviceActor = deviceIdToActor.get(msg.deviceId());
            if (Objects.nonNull(deviceActor)) {
                // 设备存在就直接回复
                msg.replyTo().tell(new DeviceManager.ResponseDeviceRegistered(deviceActor));
            } else {
                // 设备不存在就创建，保存后再回复
                log.info("Creating device actor for [{}]", msg.deviceId());
                deviceActor = getContext().spawn(Device.create(groupId, msg.deviceId()), "Device-" + msg.deviceId());

                // 一旦给定的ActorRef标识的Actor终止，就用自定义消息注册终止通知。
                // 当被监视的参与者位于使用Akka cluster时已从集群中移除的节点上时，也会发送此消息。
                getContext().watchWith(deviceActor, new DeviceTerminated(deviceActor, groupId, msg.deviceId()));
                deviceIdToActor.put(msg.deviceId(), deviceActor);
                msg.replyTo().tell(new DeviceManager.ResponseDeviceRegistered(deviceActor));
                System.out.println("=> " + groupId + " Device: " + deviceIdToActor);
            }
        } else {
            log.warn("Ignoring TrackDevice request for {}. This actor is responsible for {}.", groupId, this.groupId);
        }
        return this;
    }


    public interface GroupMessage extends Message {
    }

    /**
     * 设备停止命令
     *
     * @param device   设备引用
     * @param groupId  组编号
     * @param deviceId 设备编号
     */
    private record DeviceTerminated(ActorRef<Device.DeviceMessage> device,
                                    String groupId,
                                    String deviceId) implements GroupMessage {
    }
}
